package com.timeriver.cases.classification

import org.apache.spark.ml.{Model, Pipeline}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.ml.feature.{StandardScaler, VectorAssembler}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit, TrainValidationSplitModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import redis.clients.jedis.Jedis


/**
  * 逻辑回归训练案例：
  *   1. 读取hive数据源
  *     1.1 方式一：开启Spark对Hive的支持（推荐使用），直接使用select语句查询
  *     1.2 方式二：使用Spark JDBC方式连接（大数据下不推荐使用）
  *   2. 中间数据
  *     2.1 分析报告存储redis数据库（存储7天）
  *     2.2 中间结果存储至redis中（筛选100条，存储7天）
  *   3. 添加网格参数调优 + 交叉验证
  *   4. 使用pipeline管道训练
  *   5. 模型保存
  *   6. 算子之间特征数量删减
  *
  *   直接使用逻辑回归：
  *   预测成功率为：0.959349593495935%
  *   逻辑回归 + 所有特征标准化
  *   预测成功率为：0.9384615384615385%
  *   逻辑回归 + 所有特征标准化 + 网格搜索 + 交叉验证
  *   预测成功率为：0.9618320610687023%
  */
object LogisticAlgMySQL2Redis {
  def main(args: Array[String]): Unit = {
    val session: SparkSession = SparkSession.builder()
      .appName("逻辑回归案例")
      .master("local[6]")
      .getOrCreate()

    /** 1. 读取mysql数据 */
    val df: DataFrame = session.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/ml_datasets")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("dbtable", "breast_cancer_wisconsin")
      .option("user", "root")
      .option("password", "123456")
      .load()

    /** 过滤缺失值 */
    val value: Dataset[Row] = df.filter(!_.anyNull)

    /** 获取特征列字段数组 */
    val inputCols: Array[String] = ("clump_thickness," +
      "uniformity_of_cell_size," +
      "uniformity_of_cell_shape," +
      "marginal_adhesion," +
      "single_epithelial_cell_size," +
      "bare_nuclei," +
      "blan_chromatin," +
      "normal_nucleoli,mitoses").split(",")

    /** 注意：：：必须把标签列的列名改为label，否则在模型选择时会报label不存在 */
    val frame: DataFrame = value.withColumnRenamed("class", "label")

    /** 构建特征列向量 */
    val data: DataFrame = new VectorAssembler()
      .setInputCols(inputCols)
      .setOutputCol("features")
      .transform(frame)

    /** 数据分割 */
    val Array(train, test) = data.randomSplit(Array(0.8, 0.2))

    /** 算子1：特征归一化 */
    val scaler = new StandardScaler()
      .setInputCol("features")
      .setOutputCol("scaledFeatures")
      .setWithStd(true)
      .setWithMean(false)

    /** 算子2：创建逻辑回归模型 */
    val regression: LogisticRegression = new LogisticRegression()
      .setMaxIter(15)
      .setFeaturesCol("scaledFeatures")

    /** 构建管道模型 */
    val pipeline: Pipeline = new Pipeline().setStages(Array(scaler, regression))

    /** 构建参数网格搜索：参数调优 */
    val paramMaps: Array[ParamMap] = new ParamGridBuilder()
      .addGrid(regression.regParam, Array(0.1, 0.01))
      .addGrid(regression.fitIntercept, Array(true, false))
      .addGrid(regression.elasticNetParam, Array(0.0, 0.5, 1.0))
      .addGrid(scaler.withStd, Array(true, false))
      .addGrid(scaler.withMean, Array(true, false))
      .build()

    /** 模型评估 */
    val evaluator: BinaryClassificationEvaluator = new BinaryClassificationEvaluator()
      .setLabelCol("label")
      .setRawPredictionCol("rawPrediction")
      .setMetricName("areaUnderROC")

    /** 进行简单交叉验证 */
    val validation: TrainValidationSplit = new TrainValidationSplit()
      .setEstimator(pipeline)
      .setEvaluator(evaluator)
      .setEstimatorParamMaps(paramMaps)
      /** 80%的数据将用于训练，其余20%用于验证 */
      .setTrainRatio(0.8)
      .setParallelism(2)
      .setSeed(123)

    /** 模型训练：默认是最佳模型，bestModel是缺省值 */
    val model: TrainValidationSplitModel = validation.fit(train)

    /** 模型预测 */
    val res: DataFrame = model.transform(test)
    res.show(5, false)

    /** 获取每种参数组合的评估参数值 */
    val metrics: Array[Double] = model.validationMetrics
    metrics.foreach(x => println(s"简单交叉验证的评估参数：$x"))
    /** 获取所有的参数组合：数量根据网格搜索的参数组合确定 -> 2*3*2*2*2=48种 */
    val maps: Array[ParamMap] = model.getEstimatorParamMaps
    maps.foreach(x => println(s"${x.toString()}"))

    val accuracy: DataFrame = res.selectExpr("SUM(case when label=prediction then 1 else 0 end)/count(1) as accuracy")
    val rate: Double = accuracy.take(1)(0).getAs[Double]("accuracy")
    println(s"预测成功率为：$rate%")

    /** 将评估结果写入redis中， */
    val jedis = new Jedis("localhost", 6379, 3000)
    jedis.hset("report_key", "rate", rate.toString)

    session.stop()
  }
}
